package com.ibm.rules;

import java.util.Collection;
import java.util.Iterator;
import java.util.Map.Entry;

import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.collect.BoundType;
import com.google.common.collect.Range;
import com.google.common.collect.RangeMap;
import com.google.common.collect.TreeRangeMap;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.datamodel.timeseries.TimeSeriesSchema;
import com.ibm.rules.tree.LeafRuleTreeNode;
import com.ibm.rules.tree.RuleTreeNode;
import com.ibm.util.Pair;

/**
 * A partition on rules. Not thread safe
 * 
 * @author Yaoliang Chen
 * 
 * @param <T>
 */
public class RulePartition<T extends Comparable<T>> {
	private int tagId;
	private T leftValue;
	private BoundType leftBound;
	private boolean construction;
	private int biggestSize;

	protected RulePartition(int tagId) {
		this.ranges = TreeRangeMap.create();
		this.tagId = tagId;
		this.construction = false;
		this.biggestSize = 0;
	}

	static public <T extends Comparable<T>> RulePartition<T> create(int tagId) {
		return new RulePartition<T>(tagId);
	}

	public int getTagId() {
		return tagId;
	}

	public int partitionNumber() {
		return ranges.asMapOfRanges().size();
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	public void firstEndPoint(Pair<Comparable, Byte> endPoint) {
		try {
			leftValue = (T) endPoint.getFirst();
		} catch (ClassCastException e) {
			Throwables.propagate(e);
		}
		leftBound = ((byte) 0 == (byte) endPoint.getSecond()) ? BoundType.CLOSED
				: BoundType.OPEN;
		construction = true;
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	public void nextEndPoint(Pair<Comparable, Byte> endPoint,
			LeafRuleTreeNode node, Pair<Comparable, Byte> nextStartPoint) {
		Preconditions.checkArgument(construction,
				"trying to add a range without calling beginConstruction");
		T rightValue = null;
		BoundType rightBound = null;
		try {
			if ((byte) 0 == (byte) endPoint.getSecond()
					|| nextStartPoint == null) {
				rightValue = (T) endPoint.getFirst();
				rightBound = ((byte) 0 == (byte) endPoint.getSecond()) ? BoundType.CLOSED
						: BoundType.OPEN;
			} else {
				rightValue = (T) nextStartPoint.getFirst();
				rightBound = BoundType.OPEN;
			}
		} catch (ClassCastException e) {
			Throwables.propagate(e);
		}
		Range<T> range = Range.range(leftValue, leftBound, rightValue,
				rightBound);
		ranges.put(range, node);
		construction = false;

		int curSize = node.getRules().length;
		if (this.biggestSize < curSize)
			this.biggestSize = curSize;

		if (nextStartPoint != null)
			firstEndPoint(nextStartPoint);
	}

	public int getBiggestRangeSize() {
		return this.biggestSize;
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	public RuleTreeNode getRangeNodeByFact(Comparable[] fact) {
		try {

				T value = (T) fact[tagId];
				if (value == null) {
					return null;
				}
				return ranges.get(value);

			
			
		} catch (ClassCastException t) {
			Throwables.propagate(t);
			return null;
		}
	}
	
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public RuleTreeNode getRangeNodeByFact(ParsedDataMessage msg) {
		try {
			if (tagId >= 0) {
				T value = (T) msg.getValue(tagId);
				if (value == null) {
					return null;
				}
				return ranges.get(value);
			}else {
				int[] indexs = TimeSeriesSchema.parseTagId(tagId);
				int index = indexs[0];
				int subindex = indexs[1];
				Object[] values = (Object[])msg.getArrayValue(index);
				if (values == null) {
					return null;
				}
				T value = (T) values[subindex];
				if (value == null) {
					return null;
				}
				return ranges.get(value);
			}
			
			
		} catch (ClassCastException t) {
			Throwables.propagate(t);
			return null;
		}
	}

	@SuppressWarnings("rawtypes")
	public Collection<RuleTreeNode> getRangeNodesByRule(Rule rule) {
		try {
			Range r = rule.getRangeByTagId(tagId);
			if (r == null) {
				return ranges.asMapOfRanges().values();
			}
			return ranges.subRangeMap(r).asMapOfRanges().values();
		} catch (Exception e) {
			Throwables.propagate(e);
			return null;
		}
	}

	private RangeMap<T, RuleTreeNode> ranges;

	public Iterator<Entry<Range<T>, RuleTreeNode>> getRanges() {
		return ranges.asMapOfRanges().entrySet().iterator();
	}

	public void mergePartiton(RulePartition<T> partition) {
		ranges.putAll(partition.getRangeMap());
	}

	private RangeMap<T, RuleTreeNode> getRangeMap() {
		return ranges;
	}

}
